package com.fot.core;

import com.fot.config.QueueData;
import com.fot.config.RedisMQProperties;
import com.fot.core.listener.RedisMQListener;
import io.netty.util.internal.ThrowableUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class ConsumeWorker {
    final ScheduledExecutorService executor;

    final ScheduledExecutorService executorService;

    private final ConcurrentHashMap<String, QueueData> queueMap = new ConcurrentHashMap<>();

    private final RedissonClient redissonClient;
    private final int consumeRetryTime;
    private double currentLongingTaskCount = 0;
    private static final double perTaskConfigSize = 3000;
    private final AtomicInteger threadSize = new AtomicInteger(0);

    public ConsumeWorker(String name, final RedissonClient redissonClient,
                         RedisMQProperties configProperties) {
        this.redissonClient = redissonClient;
        this.consumeRetryTime = configProperties.getConsumeRetryTime();
        this.executor = Executors
                .newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), r -> {
                    Thread t = new Thread(r);
                    t.setName(name);
                    t.setDaemon(true);
                    return t;
                });
        this.executorService = Executors
                .newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), r -> {
                    Thread t = new Thread(r);
                    t.setName(name+ ".longPolling"+ threadSize.incrementAndGet());
                    t.setDaemon(true);
                    return t;
                });
        this.executor.scheduleWithFixedDelay(() -> {
            try {
                checkQueueInfo();
            } catch (Throwable e) {
                log.error("[sub-check] rotate check error: {}", ThrowableUtil.stackTraceToString(e));
            }
        }, 1L, 10L, TimeUnit.MILLISECONDS);
    }

    public void addListeners(String topic, List<? extends RedisMQListener<?>> listeners) {
        QueueData queueData = addQueueDataIfAbsent(topic);
        for (RedisMQListener<?> listener : listeners) {
            queueData.addListener(listener);
        }
    }

    public QueueData addQueueDataIfAbsent(String topic) {
        QueueData queueData = queueMap.get(topic);
        if (queueData != null) {
            return queueData;
        }
        queueData = new QueueData(topic);
        QueueData lastCacheData = queueMap.putIfAbsent(topic, queueData);
        if (lastCacheData == null) {
            lastCacheData = queueData;
        }
        return lastCacheData;
    }

    private void checkQueueInfo() {
        int listenerSize = queueMap.size();
        int longingTaskCount = (int) Math.ceil(listenerSize / perTaskConfigSize);
        if (longingTaskCount > currentLongingTaskCount) {
            for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
                for (QueueData queueData : queueMap.values()) {
                    if (queueData.getTaskId() == i) {
                        executorService.execute(new LongPollingRunnable(queueData));
                    }
                }
            }
            currentLongingTaskCount = longingTaskCount;
        }
    }

    class LongPollingRunnable implements Runnable {

        private final QueueData queueData;

        public LongPollingRunnable(QueueData queueData) {
            this.queueData = queueData;
        }

        @Override
        public void run() {
            try {
                RBlockingQueue<String> queue = redissonClient.getBlockingQueue(queueData.getTopic());
                String message = queue.take();
                log.info("topic: {} take message: {}", queueData.getTopic(), message);
                queueData.setMessage(message);
                queueData.notifyListener();
                executorService.execute(this);
            } catch (Exception e) {
                log.info("topic: {} take message fail: {}", queueData.getTopic(), ThrowableUtil.stackTraceToString(e));
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
                executorService.schedule(this, consumeRetryTime, TimeUnit.MILLISECONDS);
            }
        }
    }

}
